package cn.concurrence.productconsume;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 使用阻塞队列
 *
 * @author summit
 * @since 2022/4/26 11:23
 */
public class ProdConsTest3 {

	private volatile Boolean flag = true;

	private BlockingQueue<Integer> bq = new ArrayBlockingQueue<>(3);

	private AtomicInteger ai = new AtomicInteger();


	private void prods() throws InterruptedException {
		while (flag) {
			Integer data = ai.incrementAndGet();
			boolean offer = bq.offer(data, 2, TimeUnit.SECONDS);
			if (offer) {
				System.out.println(Thread.currentThread().getName() + "插入队列成功，data=>" + data);
			} else {
				System.out.println(Thread.currentThread().getName() + "插入队列" + data + "...失败");
			}
			TimeUnit.SECONDS.sleep(1);
		}

		System.out.println("product      end===========");
	}

	private void cons() throws InterruptedException {
		while (flag) {
			Integer poll = bq.poll(2, TimeUnit.SECONDS);
			if (poll != null) {
				System.out.println(Thread.currentThread().getName() + "获取数据成功，data=>" + poll);
			} else {
				System.out.println(Thread.currentThread().getName() + "获取数据失败");
			}
		}
		System.out.println("cons      end===========");
	}

	private void stop() throws InterruptedException {
		TimeUnit.SECONDS.sleep(5);
		flag = false;
		System.out.println("stop============");
	}

	public static void main(String[] args) throws InterruptedException {

		ProdConsTest3 pc = new ProdConsTest3();
		new Thread(() -> {
			try {
				pc.prods();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}

		},"生产者").start();


		new Thread(() -> {
			try {
				pc.cons();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}

		},"消费者").start();

		pc.stop();

	}
}
